package com.sunday.common.mq.kafka.study.spring.e6_Receiving_Messages_4_Batch_Manual_Ack;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.autoconfigure.kafka.ConcurrentKafkaListenerContainerFactoryConfigurer;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.ContainerCustomizer;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.config.TopicBuilder;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.kafka.listener.KafkaMessageListenerContainer;

import java.util.Map;

/**
 * https://docs.spring.io/spring-kafka/docs/current/reference/html/#receiving-messages
 * {@link org.springframework.boot.autoconfigure.kafka.KafkaAutoConfiguration}
 */
@Slf4j
@SpringBootApplication
public class Application {

    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
    }

    /**
     * https://docs.spring.io/spring-kafka/docs/current/reference/html/#batch-listeners
     */
    @Bean
    KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Object, Object>> kafkaBatchListenerContainerFactory(
            ConcurrentKafkaListenerContainerFactoryConfigurer configurer,
            ObjectProvider<ConsumerFactory<Object, Object>> kafkaConsumerFactory,
            ObjectProvider<ContainerCustomizer<Object, Object, ConcurrentMessageListenerContainer<Object, Object>>> kafkaContainerCustomizer,
            KafkaProperties properties
    ) {
        ConcurrentKafkaListenerContainerFactory<Object, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
        configurer.configure(factory, kafkaConsumerFactory
                .getIfAvailable(() -> new DefaultKafkaConsumerFactory<>(properties.buildConsumerProperties())));
        kafkaContainerCustomizer.ifAvailable(factory::setContainerCustomizer);

        factory.setBatchListener(true);

        /**
         * 手动两个值都需要设置
         * Caused by: java.lang.IllegalStateException: Consumer cannot be configured for auto commit for ackMode MANUAL
         */
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
        factory.getConsumerFactory().updateConfigs(Map.of(
                ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false
                /**
                 * Kafka 中的批量拉取是指消费者从 Kafka 代理服务器一次性获取多个消息，以提高效率并减少网络开销。
                 * 通过批量拉取，消费者能够在一次请求中获取多个消息，而不是每次只获取一个消息。
                 * 在 Kafka 中，可以通过调整以下参数来控制批量拉取的行为：
                 * fetch.min.bytes：此参数表示经纪人等待发送到消费者的最小数据量，以便触发消息传输。这有助于确保消费者能够一次性获取足够多的数据。
                 * max.poll.records：该参数表示一次最大拉取的记录数，即每次拉取消息的最大数量。
                 * 调整这些参数可以帮助优化批量拉取的行为，同时也要考虑网络吞吐量和延迟之间的平衡。
                 *
                 * 在 Kafka 中，批量拉取是由 Kafka 消费者（consumer）完成的。当消费者从 Kafka 代理服务器获取消息时，它可以配置为一次性获取多个消息，以减少网络开销并提高效率。这种一次性获取多个消息的行为被称为批量拉取。
                 * Kafka 消费者通过调用 poll() 方法来进行拉取消息，而在每次调用 poll() 方法时，消费者有机会一次性获取多个消息，而不是每次只获取一个消息。通过调整相关的配置参数，如 fetch.min.bytes 和 max.poll.records，消费者可以控制批量拉取的行为。
                 * 因此，实际上是 Kafka 消费者在内部执行批量拉取操作，以便提高整体的消息处理效率。
                 *
                 * {@link KafkaMessageListenerContainer.ListenerConsumer#pollConsumer()}
                 * consumer -> fetcher -> maxPollRecords
                 */
                , ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 4
        ));

        return factory;
    }

    @Bean
    public NewTopic topic3() {
        return TopicBuilder.name("topic3")
                .partitions(3)
                .replicas(1)
                .build();
    }

}
